package com.cayden.componets.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import org.apache.logging.log4j.util.Strings;

import java.net.InetSocketAddress;

/**
 * @author cayden
 * @date 2021-10-18 11:25
 * 单机模式
 */
public class SimpleCanalClient extends AbstractCanalClient {

    private CanalProperties canalProperties;

    public SimpleCanalClient(CanalProperties canalProperties) {
        super(canalProperties.getDestination());
        this.canalProperties = canalProperties;
        this.destination = canalProperties.getDestination();
    }

    public void run() {
        // 根据ip，直接创建链接，无HA的功能
        String destination = "example";
        if(Strings.isNotBlank(canalProperties.getDestination())){
            destination = canalProperties.getDestination();
        }
        String ip = canalProperties.getHost();
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11111),
                destination,
                "",
                "");

        //final SimpleCanalClientTest clientTest = new SimpleCanalClientTest(destination);
        this.setConnector(connector);
        this.start();
        Runtime.getRuntime().addShutdownHook(new Thread() {

            @Override
            public void run() {
                try {
                    logger.info("## stop the canal client");
                    SimpleCanalClient.this.stop();
                } catch (Throwable e) {
                    logger.warn("##something goes wrong when stopping canal:", e);
                } finally {
                    logger.info("## canal client is down.");
                }
            }
        });
    }

}
